package main

import (
	"context"
	"fmt"
	"git.medlinker.com/golang/ka"
	"git.medlinker.com/golang/xtrace"
	"github.com/gogf/gf/frame/g"
	"go.opentelemetry.io/otel/trace"
	"time"
)

var (
	brokers = []string{
		"x10.111.153.97:9092", "10.111.153.96:9092", "10.111.153.98:9092",
	}
	topic = "test-med-panel-todo"
	group = "test_consumer_wuqifei_all"
)

func main() {
	xtrace.InitJaeger("kafka_service", "localhost:6831")
	kas, err := ka.New(&ka.Config{
		Brokers: brokers,
		Consumers: []*ka.ConsumerConfig{{
			Topic: topic,
			Group: group,
		}},
	})
	if err != nil {
		panic(err)
	}
	defer kas.Close()

	{
		// 消费
		if err := kas.Consume("/UserInfo", func(ctx context.Context, e *ka.Event) error {
			spanCtx := trace.RemoteSpanContextFromContext(ctx)
			fmt.Println("consumer ka", e.Type, e.Data.MapStrStr(), e.Kv, spanCtx.SpanID.String(), spanCtx.TraceID.String())
			return nil
		}); err != nil {
			panic(err)
		}
		// 消费
		if err := kas.Consume("/UserInfo2", func(ctx context.Context, e *ka.Event) error {
			spanCtx := trace.RemoteSpanContextFromContext(ctx)
			fmt.Println("consumer ka", e.Type, e.Data.MapStrStr(), e.Kv, spanCtx.SpanID.String(), spanCtx.TraceID.String())
			return nil
		}); err != nil {
			panic(err)
		}
		consumer := kas.Consumer(topic)
		consumer.HandleError(func(err error) {
			fmt.Println("err", err)
		})
		kas.Start()
		fmt.Println("start consumer")
	}

	for {
		// 生产
		time.Sleep(time.Second)
		kas.Push(context.Background(), topic, "/UserInfo", g.Map{"ka": true, "id": time.Now().Unix()})
		kas.Push(context.Background(), topic, "/UserInfo2", g.Map{"info2": "hello", "id": time.Now().UnixNano()})
	}
}
